package com.cn.daimajiangxin.flink.sink;

import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.RollingPolicy;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;

import java.time.Duration;

public class FileSystemSinkDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<Object> stream = env.fromData("Hello", "Flink", "FileSystem", "Sink");
        RollingPolicy<Object, String> rollingPolicy = DefaultRollingPolicy.<Object, String>builder()
                .withRolloverInterval(Duration.ofMinutes(15))
                .withInactivityInterval(Duration.ofMinutes(5))
                .withMaxPartSize(MemorySize.ofMebiBytes(64))
                .build();

        // 创建文件系统Sink
        FileSink<Object> sink = FileSink
                .forRowFormat(new Path("file:///tmp/flink-output"), new SimpleStringEncoder<>())
                .withRollingPolicy(rollingPolicy)
                .build();
        // 添加Sink
        stream.sinkTo(sink);
        env.execute("File System Sink Demo");
    }
}